#include "../EventLoop.h"
#include "../Channel.h"
// #include "../Poller.h"
#include "../EPoller.h"
#include "../TimerQueue.h"
#include <signal.h>
using namespace yhaida;
__thread EventLoop *t_loopInThisThread = NULL;
const int kPollTimeMs = 1000;

static int createEventfd()
{
    int eventFd = ::eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
    if (eventFd < 0)
    {
        LOG_SYSERR << "Failed in eventfd";
    }
    return eventFd;
}

class IgnoreSigPipe
{
    public:
        IgnoreSigPipe()
        {
            ::signal(SIGPIPE,SIG_IGN);
        }
};
//默认开启
IgnoreSigPipe initObj;

EventLoop::EventLoop()
    : _looping(false),
      _quit(false),
      _callingPendingFunctors(false),
      _EventLoopThreadID(CurrentThread::tid()),
    //   _poller(new Poller(this)),
      _poller(new EPoller(this)),
      _timerQueue(new TimerQueue(this)),
      _wakeupFd(createEventfd()),
      _wakeupChannel(new Channel(this, _wakeupFd)),
      _wakeupMutex()

{
    LOG_TRACE << "EventLoop created " << this << " in thread " << _EventLoopThreadID;
    //在构造函数时t_loopThisThread已经有值则违背了one loop per thread原则
    if (getEventLoopOfCurrentThread() != NULL)
    {
        LOG_FATAL << "Another EventLoop " << t_loopInThisThread
                  << " exists in this thread " << _EventLoopThreadID;
    }
    else
    {
        t_loopInThisThread = this;
    }

    _wakeupChannel->setReadCallBackFuc(std::bind(&EventLoop::handleRead, this));
    _wakeupChannel->enableReading();
}
EventLoop::~EventLoop()
{
    assert(!_looping);
    ::close(_wakeupFd);
    t_loopInThisThread = NULL;
}
EventLoop *EventLoop::getEventLoopOfCurrentThread()
{
    return t_loopInThisThread;
}
//运行
void EventLoop::loop()
{
    assert(!_looping);
    assertInLoopThread();
    _looping = true;
    _quit = false;

    while (!_quit)
    {
        //每一轮清空一次！
        _EventLoopActiveChannels.clear();
        _pollReturnTime =  _poller->poll(&_EventLoopActiveChannels, kPollTimeMs);
        for (const auto &ch : _EventLoopActiveChannels)
        {
            // sleep(2);
            ch->handleEvent(_pollReturnTime);
        }
        // LOG_DEBUG<<"start doPendingFunctors";
        doPendingFunctors();
    }

    LOG_TRACE << "EventLoop " << this << " stop looping";
    _looping = false;
}

// assert()
void EventLoop::assertInLoopThread()
{
    if (!isInLoopThread())
    {
        abortNotInLoopThread();
    }
}
//判断 因为有可能是在工作线程
inline bool EventLoop::isInLoopThread()
{
    return CurrentThread::tid() == _EventLoopThreadID;
}

//异常
void EventLoop::abortNotInLoopThread()
{
    LOG_FATAL << "EventLoop::abortNotInLoopThread - EventLoop " << this
              << " was created in threadId_ = " << _EventLoopThreadID
              << ", current thread id = " << CurrentThread::tid();
}
void EventLoop::ChannelUpDate(Channel *channel)
{
    assert(channel->getMainEventLoop() == this);
    assertInLoopThread();
    _poller->updataChannel(channel);
}

void EventLoop::quit()
{
    _quit = true;
    if (!isInLoopThread())
    {
        wakeup();
    }
}

TimerID EventLoop::runAt(const Timestamp &time, const TimerCallback &cb)
{
    return _timerQueue->addTimer(cb, time, 0.0);
}
TimerID EventLoop::runAfter(double delay, const TimerCallback &cb)
{
    Timestamp time(addTime(Timestamp::now(), delay));
    return runAt(time, cb);
}
TimerID EventLoop::runEvery(double interval, const TimerCallback &cb)
{
    Timestamp time(addTime(Timestamp::now(), interval));
    return _timerQueue->addTimer(cb, time, interval);
}

void EventLoop::cancel(TimerID timerId)
{
    return _timerQueue->cancel(timerId);
}

void EventLoop::runInLoop(const Functor &cb)
{
    // LOG_DEBUG<<"runInLoop";
    if (isInLoopThread())
    {
        // LOG_DEBUG<<"start in this Loop";
        cb();
    }
    else
    {
        // LOG_DEBUG<<"start later";
        queueInLoop(cb);
    }
}
void EventLoop::queueInLoop(const Functor &cb)
{
    {
        MutexLockGuard guard(_wakeupMutex);
        _pendingFunctors.push_back(cb);
    }

    if (!isInLoopThread() || _callingPendingFunctors)
    {
        wakeup();
    }
}
void EventLoop::wakeup()
{
    uint64_t one = 1;
    ssize_t length = ::write(_wakeupFd, &one, sizeof(one));
    if (length != sizeof(one))
    {
        LOG_ERROR << "EventLoop::wakeup() writes " << length << " bytes instead of 8";
    }
}

void EventLoop::handleRead()
{
    uint64_t one = 1;
    ssize_t length = ::read(_wakeupFd, &one, sizeof(one));
    // sleep(5);
    if (length != sizeof(one))
    {
        LOG_ERROR << "EventLoop::wakeup() writes " << length << " bytes instead of 8";
    }
}
void EventLoop::doPendingFunctors()
{
    std::vector<Functor> functors;
    _callingPendingFunctors = true;
    {
        MutexLockGuard guard(_wakeupMutex);
        functors.swap(_pendingFunctors);
    }
    for (auto &cb : functors)
    {
        cb();
    }
    _callingPendingFunctors = false;
}
void EventLoop::removeChannel(Channel* channel)
{
    assert(channel->getMainEventLoop()==this);
    assertInLoopThread();
    _poller->removeChannel(channel);
}